package com.java.springcloud.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author dushougudu
 */
@RestController
@Slf4j
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @PostMapping("/send")
    @Transactional
    public String send(String msg) {
        kafkaTemplate.send("quickstart-events", msg).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            log.info("发送消息成功:" + topic + "-" + partition + "-" + offset);
        }, failure -> {
            log.info("发送消息失败:" + failure.getMessage());
        });

        return "success";
    }

    @PostMapping("sendInTransaction")
    public String sendInTransaction(String msg) {
        //开启事务参数后，发送都需要使用事务
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("quickstart-events", msg);
//            throw new RuntimeException("fail");
            return "success";
        });

        return "success";
    }


    @KafkaListener(topics = "mongodb.DLT", errorHandler = "consumerAwareErrorHandler")
    public void listen(ConsumerRecord<?, ?> record, Acknowledgment ack, Consumer consumer) {
        log.info("topic is {}, offset is {}, value is {} \n", record.topic(), record.offset(), record.value());
        ack.acknowledge();
    }

    @KafkaListener(topics = "mongodb", containerFactory = "filterContainerFactory")
    public void listenMongoDb(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        log.info("topic is {}, partition is {}, offset is {}, value is {} \n",
                record.topic(), record.partition(), record.offset(), record.value());
//        ack.acknowledge();
        // 模拟抛出一次一行
        throw new RuntimeException("MOCK Handle Exception Happened");
    }

}
